package com.yequan.mq.producer;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

/**
 * @author yq
 * @date 2020/09/21
 */
public class ProducerRetry {

    public static void main(String[] args) throws MQClientException, InterruptedException {
        // 声明一个生产者，需要一个自定义生产者组（后面我们会介绍这个组的概念和作用）
        DefaultMQProducer producer = new DefaultMQProducer("myTestGroup");
        // 设置集群的NameServer地址，多个地址之间以分号分隔
        producer.setNamesrvAddr("127.0.0.1:9876");
        // 如果消息发送失败就进行5次重试
        producer.setRetryTimesWhenSendFailed(2);
        //消息在broker中没有存储成功，将消息发送至另一个broker中
        producer.setRetryAnotherBrokerWhenNotStoreOK(true);
        // 启动生产者实例
        producer.start();
        // 模拟发送10条消息 到Topic为TopicTest，tag为tagA，消息内容为Hello RocketMQ +i
        for (int i = 0; i < 1; i++) {
            try {
                Message msg = new Message("TopicTest", "TagA", ("生产第" + i + "条消息").getBytes(RemotingHelper.DEFAULT_CHARSET));
                // 调用Produce的send方法发送消息
                SendResult sendResult = producer.send(msg, 1000);
                System.out.printf("%s%n", sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }
        // 发送完消息之后调用producer的shutdown（）方法关闭producer
        producer.shutdown();
    }


}
